package com.ben.chatroom.distribute;

import com.ben.chatroom.common.codec.ProtobufDecoder;
import com.ben.chatroom.common.codec.ProtobufEncoder;
import com.ben.chatroom.common.entity.ImNode;
import com.ben.chatroom.handler.ImNodeExceptionHandler;
import com.ben.chatroom.handler.ImNodeHeartBeatClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * @author
 */
@Slf4j
@Data
public class PeerSender {

  private ImNode rmNode;

  private Channel channel;
  private int reConnectCount = 0;
  private Boolean connectFlag = false;

  public PeerSender(ImNode node) {
    this.rmNode = node;
    /**
     * 客户端的是Bootstrap，服务端的则是 ServerBootstrap。
     * 都是AbstractBootstrap的子类。
     **/
    bootstrap = new Bootstrap();
    group = new NioEventLoopGroup();

  }

  /**
   * 关闭监听
   */
  private GenericFutureListener<ChannelFuture> closeListener = (ChannelFuture f) -> {
    log.info("分布式连接已经断开......{}", rmNode.toString());
    channel = null;
    connectFlag = false;
  };

  private GenericFutureListener<ChannelFuture> connectedListener = (ChannelFuture f) -> {
    EventLoop eventLoop = f.channel().eventLoop();

    if(!f.isSuccess() && ++reConnectCount < 3) {
      log.info("连接失败！在10s之后准备尝试第{}次重连！", reConnectCount);
      eventLoop.schedule(PeerSender.this::doConnect, 10, TimeUnit.SECONDS);
      this.connectFlag = false;
    }
    else {
      this.connectFlag = true;
      log.info("连接成功，{} 分布式节点连接成功:{}", new Date(), rmNode.toString());
      this.channel = f.channel();
      channel.closeFuture().addListener(closeListener);
    }

  };

  private Bootstrap bootstrap;
  private EventLoopGroup group;

  public void doConnect() {
    String host = rmNode.getHost();
    int port = rmNode.getPort();

    if(bootstrap.group() == null) {
      bootstrap
        .group(group)
        .channel(NioSocketChannel.class)
        .option(ChannelOption.SO_KEEPALIVE, true)
        .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
        .remoteAddress(host, port);

      bootstrap.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel socketChannel) {
          ChannelPipeline pipeline = channel.pipeline();
          pipeline.addLast("decoder", new ProtobufDecoder());
          pipeline.addLast("encoder", new ProtobufEncoder());
          pipeline.addLast("imNodeHeartBeatClientHandler", new ImNodeHeartBeatClientHandler());
          pipeline.addLast("exceptionHandler", new ImNodeExceptionHandler());
        }
      });

      log.info(new Date() + "开始连接分布式节点:{}", rmNode.toString());
      ChannelFuture future = bootstrap.connect();

      future.addListener(connectedListener);
    }
    else if(bootstrap.group() != null) {
      log.info(new Date() + "再一次开始连接分布式节点" + rmNode.toString());
      ChannelFuture future = bootstrap.connect();
      future.addListener(connectedListener);
    }
  }

  public boolean writeAndFlush(Object pkg) {
    if(!connectFlag) {
      log.error("分布式节点未连接" + rmNode.toString());
      return false;
    }

    channel.writeAndFlush(pkg);
    return true;
  }

  public void stopConnecting() {
    if(!connectFlag) {
      group.shutdownGracefully();
      this.connectFlag = false;
    }

  }

}
